"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : 基础功能
Case Name   : 使用pitr恢复至创建发布订阅之前
Create At   : 2022/3/21
Owner       : opentestcase026
Description :
    1.两集群开启归档，并进行基础备份
    2.在两个集群创建表
    3.两个集群均创建laber_name
    4.创建发布订阅
    5.更新数据
    6.查询数据是否更新
    7.发布端恢复至laber_name
    8.查询发布端是否存在
    9.重新创建发布
    10.更新并查询数据是否更新
    11.订阅端恢复至laber_name
    12.查询订阅是否存在
    13.重新创建订阅
    14.插入并查询是否更新
Expect      :
    1.成功
    2.成功
    3.成功
    4.成功
    5.成功
    6.更新
    7.成功
    8.不存在
    9.成功
    10.插入数据成功，不更新，复制槽不存在
    11.成功
    12.不存在
    13.成功
    14.插入数据成功，且更新成功
History     :
    Modified by opentestcase001 2022/5/5:增加等待5.5s，适配最新代码
    Modified by opentestcase004 2022/8/02:增加基础数据同步功能后订阅端有基础数据的同
    时再在订阅端修改表数据,主键冲突,关闭基础数据同步
    Modified by opentestcase004 2022/8/12:清理环境重启库添加断言
    Modified by opentestcase004 2022/9/23:公共函数参数限制，优化用例对订阅端备机的重
    建方式
    Modified by opentestcase012 2022/10/11:优化用例和断言,添加查看数据库状态
    Modified by opentestcase012 2022/11/9 优化用例，build之后添加等待主备同步
"""
import unittest
import os
import time
from yat.test import macro
from yat.test import Node
from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Common import Common
from testcase.utils.Constant import Constant

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != Primary_SH.get_node_num(), '非1+2环境不执行')
class Pubsubclass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("-----------this is setup-----------")
        self.log.info(f"-----{os.path.basename(__file__)} start-----")
        self.pri_userdb_pub = Node(node='PrimaryDbUser')
        self.pri_userdb_sub = Node(node='remote1_PrimaryDbUser')
        self.constant = Constant()
        self.commsh_pub = CommonSH('PrimaryDbUser')
        self.commsh_sub = CommonSH('remote1_PrimaryDbUser')
        self.com_pub = Common()
        self.com_sub = Common('remote1_PrimaryDbUser')
        self.tb_name1 = 'tb_pubsub_case073_1'
        self.tb_name2 = 'tb_pubsub_case073_2'
        self.tbs_name = 'tps_pubsub_case073'
        self.schema_name1 = 's_pubsub_case073_1'
        self.subname1 = "sub_case073_1"
        self.pubname1 = "pub_case073_1"
        self.parent_path_pub = os.path.dirname(macro.DB_INSTANCE_PATH)
        self.parent_path_sub = os.path.dirname(macro.DB_INSTANCE_PATH_REMOTE1)
        self.hba_pub = os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')
        self.hba_sub = os.path.join(macro.DB_INSTANCE_PATH_REMOTE1,
                                    'pg_hba.conf')
        self.hba_bak_pub = os.path.join(
            os.path.dirname(macro.DB_INSTANCE_PATH), 'pg_hba.conf')
        self.hba_bak_sub = os.path.join(
            os.path.dirname(macro.DB_INSTANCE_PATH_REMOTE1), 'pg_hba.conf')
        self.port = str(int(self.pri_userdb_pub.db_port) + 1)
        self.wal_level = self.com_pub.show_param("wal_level")
        self.user_param_pub = f'-U {self.pri_userdb_pub.db_user} ' \
            f'-W {self.pri_userdb_pub.db_password}'
        self.user_param_sub = f'-U {self.pri_userdb_sub.db_user} ' \
            f'-W {self.pri_userdb_sub.db_password}'

        self.backup_path_pub = os.path.join(self.parent_path_pub,
                                            'back_pubsub_case073')
        self.backup_path_sub = os.path.join(self.parent_path_sub,
                                            'back_pubsub_case073')
        self.comsh_sta_pub = []
        self.nodelist_pub = ['Standby1DbUser', 'Standby2DbUser']
        self.num_pub = int(self.commsh_pub.get_node_num()) - 1
        self.log.info(str(self.num_pub))
        for i in range(self.num_pub):
            self.comsh_sta_pub.append(CommonSH(self.nodelist_pub[i]))
        self.comsh_sta_sub = []
        self.nodelist_sub = ['remote1_Standby1DbUser',
                             'remote1_Standby2DbUser']
        for i in range(len(self.nodelist_sub)):
            self.comsh_sta_sub.append(CommonSH(self.nodelist_sub[i]))
        self.tbs_path_pub = os.path.join(self.parent_path_pub, self.tbs_name)
        self.tbs_path_sub = os.path.join(self.parent_path_sub, self.tbs_name)
        self.archive_path_pub = os.path.join(self.parent_path_pub, 'archive')
        self.archive_path_sub = os.path.join(self.parent_path_sub, 'archive')
        self.recovery_path_pub = os.path.join(macro.DB_INSTANCE_PATH,
                                              'recovery.conf')
        self.recovery_path_sub = os.path.join(macro.DB_INSTANCE_PATH_REMOTE1,
                                              'recovery.conf')
        self.archive_mode_pub = self.com_pub.show_param("archive_mode")
        self.enable_slot_log_pub = self.com_pub.show_param("enable_slot_log")
        self.archive_mode_sub = self.com_sub.show_param(
            "archive_mode", macro.DB_ENV_PATH_REMOTE1)
        self.enable_slot_log_sub = self.com_sub.show_param(
            "enable_slot_log", macro.DB_ENV_PATH_REMOTE1)
        self.hostname_pub = self.pri_userdb_pub.sh('hostname').result()
        self.hostname_sub = self.pri_userdb_sub.sh('hostname').result()

        text = '-----预置条件1:备份配置文件pg_hba.conf expect:状态正常-----'
        self.log.info(text)
        cmd = f'cp {self.hba_pub} {self.hba_bak_pub};'
        self.log.info(cmd)
        msg1 = self.com_pub.get_sh_result(self.pri_userdb_pub, cmd)
        cmd = f'cp {self.hba_sub} {self.hba_bak_sub};'
        self.log.info(cmd)
        msg2 = self.com_pub.get_sh_result(self.pri_userdb_sub, cmd)
        self.assertEqual('', msg1 and msg2, '执行失败:' + text)

        text = '-----预置条件2:查看发布端和订阅端的数据库状态 expect:状态正常-----'
        self.log.info(text)
        status = self.commsh_pub.get_db_cluster_status()
        self.log.info(status)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status, '执行失败:' + text)
        status = self.commsh_sub.get_db_cluster_status(
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status, '执行失败:' + text)

        text = '-----预置条件3:修改发布端pg_hba文件和配置参数wal_level,重启数据库 expect:成功-----'
        self.log.info(text)
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)
        if 'logical' != self.wal_level:
            result = self.commsh_pub.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical')
            self.assertTrue(result, '执行失败:' + text)
            result = self.commsh_pub.restart_db_cluster(False)
            self.assertTrue(result, '执行失败:' + text)
            status = self.commsh_pub.get_db_cluster_status()
            self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status,
                          '执行失败:' + text)

        text = '-----预置条件4:订阅端修改pg_hba文件 expect:成功-----'
        self.log.info(text)
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)

    def test_pubsub(self):
        text = '--step1:两集群开启归档，并进行基础备份 expect:成功--'
        self.log.info(text)
        self.log.info("发布端:\n")
        cmd = f"mkdir {self.backup_path_pub};mkdir {self.archive_path_pub};"
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(result)
        result = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, 'archive_mode=on',
            self.hostname_pub)
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f"archive_command = 'cp %p {self.archive_path_pub}/%f'",
            self.hostname_pub)
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, 'enable_slot_log=on')
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_pub.restart_db_cluster(False)
        self.assertTrue(result, '执行失败:' + text)
        status = self.commsh_pub.get_db_cluster_status()
        self.log.info(status)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status, '执行失败:' + text)

        time.sleep(10)
        result = self.commsh_pub.exec_gs_basebackup(
            self.backup_path_pub, self.pri_userdb_pub.ssh_host,
            self.pri_userdb_pub.db_port, )
        self.assertTrue(result, '执行失败:' + text)
        cmd = f"chmod 700 {self.backup_path_pub}"
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(result)
        self.log.info("订阅端：\n")
        cmd = f"mkdir {self.backup_path_sub};mkdir {self.archive_path_sub};"
        self.log.info(cmd)
        result = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(result)
        result = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, 'archive_mode=on',
            self.hostname_sub, dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f"archive_command = 'cp %p {self.archive_path_sub}/%f'",
            self.hostname_sub, dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, 'enable_slot_log=on',
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_sub.restart_db_cluster(
            False, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(result, '执行失败:' + text)
        status = self.commsh_sub.get_db_cluster_status(
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status, '执行失败:' + text)

        time.sleep(10)
        result = self.commsh_sub.exec_gs_basebackup(
            self.backup_path_sub, self.pri_userdb_sub.ssh_host,
            self.pri_userdb_sub.db_port,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(result, '执行失败:' + text)
        cmd = f"chmod  700 {self.backup_path_sub}"
        self.log.info(cmd)
        result = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(result)

        text = '--step2:两个集群均创建表 expect:成功--'
        self.log.info(text)
        sql1 = f"drop tablespace if exists {self.tbs_name};" \
            f"create tablespace {self.tbs_name} " \
            f" location '{self.tbs_path_pub}';"
        sql2 = f"drop tablespace if exists {self.tbs_name};" \
            f"create tablespace {self.tbs_name} " \
            f" location '{self.tbs_path_sub}';"
        sql = f"drop schema if exists {self.schema_name1};" \
            f"create schema {self.schema_name1};" \
            f"drop table if exists {self.schema_name1}.{self.tb_name1};" \
            f"create table {self.schema_name1}.{self.tb_name1}" \
            f"(id1 int primary key, id2 int, id3 int) " \
            f"tablespace {self.tbs_name};" \
            f"drop table if exists {self.tb_name2};" \
            f"create table {self.tb_name2}(id int primary key " \
            f"constraint id_nn not null,use_filename varchar2(20)," \
            f"filename varchar2(255),text varchar2(2000))" \
            f"PARTITION BY RANGE(id)" \
            f"(partition p1 values less than(30), " \
            f"partition p2 values less than(60)," \
            f"partition p3 values less than(90)," \
            f" partition p4 values less than(maxvalue));"
        result = self.commsh_pub.execut_db_sql(
            sql1 + sql, sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.TABLE_CREATE_SUCCESS),
                         5, '执行失败:' + text)
        result = self.commsh_sub.execut_db_sql(sql2 + sql, self.user_param_sub,
                                               None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.TABLE_CREATE_SUCCESS),
                         5, '执行失败:' + text)

        text = '--step3:两个集群均创建laber_name expect:成功--'
        self.log.info(text)
        sql = "select pg_switch_xlog();" \
              "select pg_create_restore_point('point1');"
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn('pg_create_restore_point', result, '执行失败' + text)
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn('pg_create_restore_point', result, '执行失败' + text)

        text = '--step4:创建发布订阅 expect:成功--'
        self.log.info(text)
        sql = f"drop publication if exists {self.pubname1};" \
            f"create publication {self.pubname1} for all tables;"
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.create_pub_succ_msg, result,
                      '执行失败:' + text)
        self.assertNotIn(self.constant.SQL_WRONG_MSG[1], result,
                         '执行失败:' + text)
        result = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn('', result, '执行失败:' + text)
        sql = f"drop subscription if exists {self.subname1};" \
            f"create subscription {self.subname1} connection " \
            f"'host={self.pri_userdb_pub.db_host} " \
            f"port={self.port} " \
            f"user={self.pri_userdb_pub.db_user} " \
            f"dbname={self.pri_userdb_pub.db_name} " \
            f"password={self.pri_userdb_pub.ssh_password}' " \
            f"publication {self.pubname1} with (copy_data=false);"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.create_sub_succ_msg), 1,
                         '执行失败:' + text)

        text = '--step5:更新数据 expect:成功--'
        self.log.info(text)
        sql_insert = f"insert into {self.schema_name1}.{self.tb_name1} " \
            f"values(generate_series(1,100), 1, 1);" \
            f"insert into {self.tb_name2} values(1, 'first', '%一', '')," \
            f"(60, 'first', '%二', ''),(90, 'first', '%三', ''); " \
            f"select pg_sleep(5.5);"
        result = self.commsh_pub.execut_db_sql(sql_insert,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.INSERT_SUCCESS_MSG),
                         2, '执行失败' + text)

        text = "--step6:查询数据是否更新 expect:更新--"
        self.log.info(text)
        sql_select = f"select count(*) from " \
            f"{self.schema_name1}.{self.tb_name1};" \
            f"select * from {self.tb_name2};"
        result = self.commsh_sub.execut_db_sql(sql_select,
                                               self.user_param_sub,
                                               None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn('3 rows', result, '执行失败' + text)
        self.assertIn('1 | first        | %一     |', result, '执行失败' +
                      text)
        result_a = self.commsh_pub.execut_db_sql(sql_select,
                                                 sql_type=self.user_param_pub)
        self.log.info(result_a)
        self.assertEqual(result_a, result, '执行失败' + text)

        text = "--step7:发布端恢复至laber_name expect:更新--"
        self.log.info(text)
        result = self.commsh_pub.stop_db_cluster()
        self.assertTrue(result, '执行失败:' + text)
        cmd = f"mv {macro.DB_INSTANCE_PATH} {macro.DB_INSTANCE_PATH}_bak;" \
            f"mv {self.backup_path_pub} {macro.DB_INSTANCE_PATH};" \
            f"ls -al {self.archive_path_pub}"
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(result)
        cmd = f"touch {self.recovery_path_pub};" \
            f"echo \"restore_command = " \
            f"'cp {self.archive_path_pub}/%f %p'\" > " \
            f"{self.recovery_path_pub};" \
            f"echo \"recovery_target_name='point1'\" >> " \
            f"{self.recovery_path_pub};" \
            f"cat {self.recovery_path_pub};"
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(result)
        result = self.commsh_pub.start_db_cluster(True)
        flg = self.constant.START_SUCCESS_MSG in result or 'Degrade' in result
        self.assertTrue(flg, '执行失败:' + text)
        time.sleep(20)
        cmd_resume = "select pg_is_in_recovery();" \
                     "select pg_xlog_replay_resume();select pg_sleep(10);" \
                     "select pg_is_in_recovery();"
        result = self.commsh_pub.execut_db_sql(cmd_resume)
        self.log.info(result)
        self.assertEqual(19, len(result.splitlines()), '执行失败:' + text)
        self.assertIn('f', result.splitlines()[-2], '执行失败:' + text)
        for i in range(self.num_pub):
            result = self.comsh_sta_pub[i].build_standby("-t 3600")
            self.assertIn(self.constant.BUILD_SUCCESS_MSG,
                          result, '执行失败:' + text)

        text = '-----校验备机是否完成数据同步-----'
        self.log.info(text)
        check_msg = self.commsh_pub.check_location_consistency(
            'primary', self.commsh_pub.get_node_num())
        self.assertTrue(check_msg, '执行失败:' + text)
        status = self.commsh_pub.get_db_cluster_status()
        self.log.info(status)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status, '执行失败:' + text)
        cmd = f"ls -al  {macro.DB_INSTANCE_PATH}"
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(result)
        result = self.commsh_pub.execut_db_sql(sql_select,
                                               sql_type=self.user_param_pub)
        self.log.info(result)

        text = '--step8:查询发布端是否存在 expect:不存在--'
        self.log.info(text)
        sql = "select * from pg_publication;"
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertNotIn(self.pubname1, result, '执行失败' + text)

        text = '--step9:重新创建发布 expect:成功--'
        self.log.info(text)
        sql = f"create publication {self.pubname1} for all tables;"
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.create_pub_succ_msg, result,
                      '执行失败:' + text)
        self.assertNotIn(self.constant.SQL_WRONG_MSG[1], result,
                         '执行失败:' + text)

        text = '--step10:更新并查询数据是否更新 expect:插入数据成功，不更新，' \
               '复制槽不存在--'
        self.log.info(text)
        sql_insert = f"insert into {self.schema_name1}.{self.tb_name1} " \
            f"values(200, 1, 1);" \
            f"insert into {self.tb_name2} values" \
            f"(200, 'first', '%一', 'new');" \
            f"select pg_sleep(30);" \
            f"select active from pg_replication_slots where " \
            f"slot_name='{self.subname1}';"
        result = self.commsh_pub.execut_db_sql(sql_insert,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.INSERT_SUCCESS_MSG),
                         2, '执行失败' + text)
        self.assertEqual(10, len(result.splitlines()), '执行失败:' + text)
        self.assertIn('0 rows', result.splitlines()[-1].strip(),
                      '执行失败' + text)
        result = self.commsh_sub.execut_db_sql(sql_select,
                                               self.user_param_sub,
                                               None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn('3 rows', result, '执行失败' + text)
        self.assertIn('1 row', result, '执行失败' + text)
        self.assertNotIn('101', result, '执行失败' + text)
        self.assertNotIn('200 | first        | %一     | new',
                         result, '执行失败' + text)

        text = "--step11:订阅端恢复至laber_name expect:更新--"
        self.log.info(text)
        result = self.commsh_sub.stop_db_cluster(
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(result, '执行失败:' + text)
        cmd = f"mv {macro.DB_INSTANCE_PATH_REMOTE1} " \
            f"{macro.DB_INSTANCE_PATH_REMOTE1}_bak;" \
            f"mv {self.backup_path_sub} {macro.DB_INSTANCE_PATH_REMOTE1};"
        self.log.info(cmd)
        result = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(result)
        cmd = f"touch {self.recovery_path_sub};" \
            f"echo \"restore_command = " \
            f"'cp {self.archive_path_sub}/%f %p'\" > " \
            f"{self.recovery_path_sub};" \
            f"echo \"recovery_target_name='point1'\" >> " \
            f"{self.recovery_path_sub};" \
            f"cat {self.recovery_path_sub};"
        self.log.info(cmd)
        result = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(result)
        result = self.commsh_sub.start_db_cluster(
            True, env_path=macro.DB_ENV_PATH_REMOTE1)
        flg = self.constant.START_SUCCESS_MSG in result or 'Degrade' in result
        self.assertTrue(flg, '执行失败:' + text)
        time.sleep(20)
        result = self.commsh_sub.execut_db_sql(cmd_resume,
                                               self.user_param_sub,
                                               None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(19, len(result.splitlines()), '执行失败:' + text)
        self.assertIn('f', result.splitlines()[-2], '执行失败:' + text)
        result = self.commsh_sub.get_standby_and_build(
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.BUILD_SUCCESS_MSG, str(result),
                      '执行失败:' + text)
        status = self.commsh_sub.get_db_cluster_status(
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status, '执行失败:' + text)

        text = '--step12:查询订阅是否存在 expect:不存在--'
        self.log.info(text)
        sql = "select * from pg_subscription;"
        result = self.commsh_sub.execut_db_sql(sql,
                                               self.user_param_sub,
                                               None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertNotIn(self.subname1, result, '执行失败:' + text)

        text = '--step13:重新创建订阅 expect:成功--'
        self.log.info(text)
        result = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn('', result, '执行失败:' + text)
        sql = f"create subscription {self.subname1} connection " \
            f"'host={self.pri_userdb_pub.db_host} " \
            f"port={self.port} " \
            f"user={self.pri_userdb_pub.db_user} " \
            f"dbname={self.pri_userdb_pub.db_name} " \
            f"password={self.pri_userdb_pub.ssh_password}' " \
            f"publication {self.pubname1} with (copy_data=false);"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.create_sub_succ_msg), 1,
                         '执行失败:' + text)

        text = '--step14:插入并查询是否更新 expect:插入数据成功，且更新成功--'
        self.log.info(text)
        sql_insert = f"insert into {self.schema_name1}.{self.tb_name1} " \
            f"values(222, 1, 1);" \
            f"insert into {self.tb_name2} values(222, 'first', '%一', 'new');" \
            f"select pg_sleep(30);" \
            f"select * from pg_replication_slots;"
        result = self.commsh_pub.execut_db_sql(sql_insert,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.INSERT_SUCCESS_MSG),
                         2, '执行失败' + text)
        sql_select = f"select * from " \
            f"{self.schema_name1}.{self.tb_name1};" \
            f"select * from {self.tb_name2};"
        result = self.commsh_sub.execut_db_sql(sql_select,
                                               self.user_param_sub,
                                               None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('1 row'), 2, '执行失败' + text)
        self.assertIn('222 | first        | %一     | new',
                      result, '执行失败' + text)
        self.assertIn('222 |   1 |   1', result, '执行失败' + text)

    def tearDown(self):
        self.log.info('-----清理环境-----')
        text1 = '-----发布端数据库重建备机查看集群状态 expect:成功-----'
        self.log.info(text1)
        status1_pub = self.commsh_pub.get_db_cluster_status('status')
        self.log.info(status1_pub)
        if not status1_pub:
            check_cmd = \
                f'if [[ -d {macro.DB_INSTANCE_PATH} && -d ' \
                    f'{macro.DB_INSTANCE_PATH}_bak ]]; ' \
                    f'then echo "exists"; else echo "not exists"; fi'
            self.log.info(check_cmd)
            result = self.pri_userdb_pub.sh(check_cmd).result()
            if result == 'exists':
                cmd = f"rm -rf {macro.DB_INSTANCE_PATH_REMOTE1};"
                result = self.pri_userdb_pub.sh(cmd).result()
                self.log.info(result)
            cmd = f"mv {macro.DB_INSTANCE_PATH}_bak {macro.DB_INSTANCE_PATH};"
            self.log.info(cmd)
            result = self.pri_userdb_pub.sh(cmd).result()
            self.log.info(result)
            self.commsh_pub.restart_db_cluster()
            for i in range(self.num_pub):
                self.comsh_sta_pub[i].build_standby("-t 3600")
            status1_pub = self.commsh_pub.get_db_cluster_status('status')
            self.log.info(status1_pub)

        text2 = '-----订阅端数据库重建备机查看集群状态 expect:成功-----'
        self.log.info(text2)
        status1_sub = self.commsh_sub.get_db_cluster_status(
            'status', env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status1_sub)
        if not status1_sub:
            check_cmd = \
                f'if [[ -d {macro.DB_INSTANCE_PATH} && -d ' \
                    f'{macro.DB_INSTANCE_PATH}_bak ]]; ' \
                    f'then echo "exists"; else echo "not exists"; fi'
            self.log.info(check_cmd)
            result = self.pri_userdb_sub.sh(check_cmd).result()
            if result == 'exists':
                cmd = f"rm -rf {macro.DB_INSTANCE_PATH_REMOTE1};"
                result = self.pri_userdb_sub.sh(cmd).result()
                self.log.info(result)
            cmd = f"mv {macro.DB_INSTANCE_PATH_REMOTE1}_bak " \
                f"{macro.DB_INSTANCE_PATH_REMOTE1};"
            self.log.info(cmd)
            result = self.pri_userdb_sub.sh(cmd).result()
            self.log.info(result)
            self.commsh_sub.restart_db_cluster(True,
                                               macro.DB_ENV_PATH_REMOTE1)
            build_result = self.commsh_sub.get_standby_and_build(
                dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
                env_path=macro.DB_ENV_PATH_REMOTE1)
            self.log.info(build_result)
            status1_sub = self.commsh_sub.get_db_cluster_status(
                'status', env_path=macro.DB_ENV_PATH_REMOTE1)
            self.log.info(status1_sub)

        text3 = '-----删除发布订阅 expect:成功-----'
        self.log.info(text3)
        sql = f"drop subscription if exists {self.subname1};"
        drop_sub_result = self.commsh_sub.execut_db_sql(
            sql, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_sub_result)
        sql = f"drop publication if exists {self.pubname1};" \
            f"purge recyclebin"
        drop_pub_result = self.commsh_pub.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(drop_pub_result)

        text4 = '-----订阅端和发布端删除表,schema和表空间 expect:成功-----'
        self.log.info(text4)
        sql = f"drop table if exists {self.schema_name1}.{self.tb_name1};" \
            f"drop table if exists {self.tb_name2};" \
            f"drop schema {self.schema_name1};" \
            f"drop tablespace {self.tbs_name};"
        sql_drop_sub = self.commsh_sub.execut_db_sql(sql, self.user_param_sub,
                                                     None,
                                                     macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sql_drop_sub)
        sql_drop_pub = self.commsh_pub.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(sql_drop_pub)

        text5 = '-----订阅端和发布端恢复pg_hba.conf文件 expect:成功-----'
        self.log.info(text5)
        cmd = f'if [[ -e {self.hba_pub} && -e {self.hba_bak_pub} ]]; ' \
            f'then rm -rf {self.hba_pub}; ' \
            f'mv {self.hba_bak_pub} {self.hba_pub}; ' \
            f'else echo not exits; fi'
        hba_msg_pub = self.com_pub.get_sh_result(self.pri_userdb_pub, cmd)
        cmd = f'if [[ -e {self.hba_sub} && -e {self.hba_bak_sub} ]]; ' \
            f'then rm -rf {self.hba_sub}; ' \
            f'mv {self.hba_bak_sub} {self.hba_sub}; ' \
            f'else echo not exits; fi'
        hba_msg_sub = self.com_pub.get_sh_result(self.pri_userdb_sub, cmd)

        text6 = '-----订阅端和发布端删除备份文件 expect:成功-----'
        self.log.info(text6)
        cmd = f"rm -rf {self.backup_path_pub};" \
            f"rm -rf {macro.DB_INSTANCE_PATH}_bak;" \
            f"rm -rf {self.tbs_path_pub};" \
            f"rm -rf {self.archive_path_pub};" \
            f"rm -rf {self.recovery_path_pub}"
        self.log.info(cmd)
        rm_msg_pub = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(rm_msg_pub)
        cmd = f"rm -rf {self.tbs_path_sub};" \
            f"rm -rf {self.backup_path_sub};" \
            f"rm -rf {macro.DB_INSTANCE_PATH_REMOTE1}_bak;" \
            f"rm -rf {self.archive_path_sub};" \
            f"rm -rf {self.recovery_path_sub}"
        self.log.info(cmd)
        rm_msg_sub = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(rm_msg_sub)

        text7 = '-----订阅端和发布端恢复参数配置 expect:成功-----'
        self.log.info(text7)
        guc_wal_pub = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.wal_level}')
        guc_archive_pub = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'archive_mode={self.archive_mode_pub}',
            self.hostname_pub)
        guc_slot_pub = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'enable_slot_log={self.enable_slot_log_pub}')
        guc_archive_sub = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'archive_mode={self.archive_mode_sub}',
            self.hostname_sub, dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        guc_slot_sub = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'enable_slot_log={self.enable_slot_log_sub}',
            self.hostname_sub, dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)

        text8 = '-----重启数据库查看数据库状态 expect:成功-----'
        self.log.info(text8)
        res_msg_pub = self.commsh_pub.restart_db_cluster(False)
        status_pub = self.commsh_pub.get_db_cluster_status()
        self.log.info(status_pub)
        res_msg_sub = self.commsh_sub.restart_db_cluster(
            env_path=macro.DB_ENV_PATH_REMOTE1)
        status_sub = self.commsh_sub.get_db_cluster_status(
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status_sub)

        self.assertTrue(status1_pub, '执行失败:' + text1)
        self.assertTrue(status1_sub, '执行失败:' + text2)
        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_result,
                      '执行失败' + text3)
        self.assertIn(self.constant.drop_pub_succ_msg, drop_pub_result,
                      '执行失败' + text3)
        self.assertIn('PURGE RECYCLEBIN', drop_pub_result, '执行失败' + text3)
        self.assertEqual(3, sql_drop_sub.count(
            self.constant.TABLE_DROP_SUCCESS), '执行失败' + text4)
        self.assertEqual(3, sql_drop_pub.count(
            self.constant.TABLE_DROP_SUCCESS), '执行失败' + text4)
        self.assertIn(self.constant.DROP_SCHEMA_SUCCESS_MSG, sql_drop_sub,
                      '执行失败' + text4)
        self.assertIn(self.constant.DROP_SCHEMA_SUCCESS_MSG, sql_drop_pub,
                      '执行失败' + text4)
        self.assertIn(self.constant.TABLESPCE_DROP_SUCCESS, sql_drop_sub,
                      '执行失败' + text4)
        self.assertIn(self.constant.TABLESPCE_DROP_SUCCESS, sql_drop_pub,
                      '执行失败' + text4)
        self.assertEqual('', hba_msg_pub and hba_msg_sub, '执行失败:' + text5)
        self.assertEqual('', rm_msg_pub and rm_msg_sub, '执行失败:' + text6)
        self.assertTrue(guc_wal_pub, '执行失败:' + text7)
        self.assertTrue(guc_archive_pub, '执行失败:' + text7)
        self.assertTrue(guc_slot_pub, '执行失败:' + text7)
        self.assertTrue(guc_archive_sub, '执行失败:' + text7)
        self.assertTrue(guc_slot_sub, '执行失败:' + text7)
        self.assertTrue(res_msg_pub, '执行失败:' + text8)
        self.assertTrue(res_msg_sub, '执行失败:' + text8)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status_pub,
                      '执行失败:' + text8)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status_sub,
                      '执行失败:' + text8)
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
